package com.kygeng.datamonitor.consumer;

import com.alibaba.fastjson.JSONObject;
import com.kygeng.common.config.KafkaConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import java.util.List;


@Component
@Slf4j
public class KafkaCompanyDataConsumer {
    @KafkaListener(topics = KafkaConfig.TOPIC_COMPANY, groupId = KafkaConfig.GROUP_ID_COMPANY)
    public void topicCompanyConsumer(List<String> messages, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
        for (String message : messages) {
            JSONObject messageJSON = JSONObject.parseObject(message);

            String string = messageJSON.getString("type");

            if (string.equals("insert")) {
                System.out.println("新增了：" + messageJSON.getString("data"));
            } else if (string.equals("update")) {
                System.out.println("更新了：" + messageJSON.getString("data"));
            } else if (string.equals("delete")) {
                System.out.println("删除了：" + messageJSON.getString("data"));
            }
        }
    }

    public void getDataFromCompany(String companyCode){

    }

    public void insetDataToCompanyEs(String companyCode){

    }

}
